package com.chinasoft.dataclean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;

import java.io.IOException;
import java.sql.Timestamp;

// 数据清洗
public class BehaviorDataCleaning {
    // 一般来说输入输出路径需要动态设置
    private static String in = "/user/hive/warehouse/ods.db/behavior.csv";
    private static String out = "/user/hive/warehouse/dwd.db/behavior";

    public static void main(String[] args) throws IOException {

        // 创建Spark配置，设置应用名称和运行模式
        SparkConf conf = new SparkConf()
                .setAppName("BehaviorDataCleaning")
                .setMaster("local[*]");

        // 创建Spark上下文，这是Spark应用程序的入口点
        JavaSparkContext context = new JavaSparkContext(conf);

        // 检测输出文件是否存在，若存在，则删除
        checkExistFile(context.hadoopConfiguration(), new Path(out));

        // 从指定路径读取文本文件，创建RDD
        // JavaRDD是Apache Spark中最基本的数据抽象，它是 Spark 对分布式数据集的表示。
        JavaRDD<String> inRDD = context.textFile(in);

        // 获取RDD中的第一条记录
//        List<String> take = inRDD.take(1);

        // 将获取的记录重新并行化为RDD
//        JavaRDD<String> parallelizeRDD = context.parallelize(take);

        // 使用map转换对每条记录进行处理
        // 在 Spark 中，操作分为两类：
        // 转换操作（Transformations）: 如 map, filter 等，是惰性求值的，不会立即执行
        // 行动操作（Actions）: 如 collect, count, take 等，会触发实际计算

        // collect() 是一个行动操作，它的主要功能包括：
        // 触发计算执行: 强制执行之前定义的所有转换操作
        // 收集结果: 将分布式 RDD 中的所有数据收集到驱动程序（Driver）中
        // 返回结果: 返回一个包含所有元素的数组
        // 如果没有 collect()，map 转换操作永远不会执行

        // 使用 collect() 需要注意：
        // 内存限制: 会将所有数据加载到驱动程序内存中，只适用于小数据集
        // 网络传输: 需要从所有工作节点传输数据到驱动程序
        // 适用场景: 主要用于调试或小规模数据处理
        // 对于生产环境的大数据集，通常使用 foreach() 或将结果保存到外部存储系统。
        JavaRDD<String> mapRDD = inRDD.map(new Function<String, String>() {
            @Override
            public String call(String s) throws Exception {

                String[] fields = s.split(",");

                // 过滤异常长度的数据
                if (fields.length == 5) {
                    // 获取数据字段
                    String userId = fields[0];
                    String itemId = fields[1];
                    String categoryId = fields[2];
                    String type = fields[3];
                    String timestamp = fields[4];

                    // 清洗时间格式
                    // Long.valueOf()用Long.parseLong()代替更高效，并且不需要转换类型(仍然是字符串)
                    Timestamp tmpTime = new Timestamp(Long.parseLong(timestamp) * 1000L);
                    // System.out.println("tmpTime.toString() = " + tmpTime); // 输出测试

                    String times = tmpTime.toString().substring(0, 19); // 去掉时间最后的.0

                    String day = tmpTime.toString().substring(0, 10); // 清洗日期
                    String hour = tmpTime.toString().substring(11, 13); // 清洗小时

                    // 清洗时间范围
                    if(day.compareTo("2017-11-25") >= 0 && day.compareTo("2017-12-03") <= 0){
                        return userId + "," + itemId + "," + categoryId + "," + type + "," + times + "," + day + "," + hour;
                    }else{
                        return null;
                    }
                } else {
                    return null;
                }
            }
        }).filter(row -> null != row); // 清洗null值

        mapRDD.saveAsTextFile(out);
        context.close();
    }

    // 检测文件是否存在
    private static void checkExistFile(Configuration conf, Path path) throws IOException {
        FileSystem fileSystem = FileSystem.get(conf);
        if(fileSystem.exists(path)){
            fileSystem.delete(path, true);
            fileSystem.close();
        }
    }
}
